package com.bang.framework.mq.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.UUID;

/**
 * @ClassName: MqSendUtils
 * @author: BANG
 * @Date: 2019/12/10 09:33
 * @Version: 1.0
 * @Modified By:
 * @Desc: 普通消息发送工具类, 不保证消息可靠性
 */
@Component
@Slf4j
public class MqSendUtils implements RabbitTemplate.ReturnCallback, RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    public <T> void send(T message, String routerKey) throws Exception {
        log.info("写入消息队列 >>> routerKey：{}，消息内容：{}", routerKey, JSON.toJSONString(message));
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        //构建回调id为uuid
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        //发送消息到消息队列
        rabbitTemplate.convertAndSend(routerKey, message, correlationId);
    }

    /**
     * @param message   消息内容
     * @param exchange  队列交换
     * @param routerKey 队列交换绑定的路由键
     * @throws Exception
     */
    public <T> void send(T message, String exchange, String routerKey) throws Exception {
        log.info("写入消息队列 >>> exchange：{} routerKey：{}，消息内容：{}", exchange, routerKey, JSON.toJSONString(message));
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        //构建回调id为uuid
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        //发送消息到消息队列
        rabbitTemplate.convertAndSend(exchange, routerKey, message, correlationId);
    }


    /**
     * @param messageContent 消息内容
     * @param exchange       队列交换
     * @param routerKey      队列交换绑定的路由键
     * @param delayTimes     延迟时长，单位：毫秒
     * @throws Exception
     */
    public <T> void sendTtlMessage(T messageContent, String exchange, String routerKey, final long delayTimes) throws Exception {
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        //构建回调id为uuid
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        //构建回调id为uuid
        if (!StringUtils.isEmpty(exchange)) {
            log.info("延迟：{}毫秒写入消息队列 >>>> exchange：{} routerKey：{}，消息内容：{}", delayTimes, exchange, routerKey, JSON.toJSONString(messageContent));
            // 执行发送消息到指定队列
            rabbitTemplate.convertAndSend(exchange, routerKey, messageContent, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                    return message;
                }
            }, correlationId);
        } else {
            log.error("未找到队列消息：{}，所属的交换机", exchange);
        }
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.error("correlationData: {}", correlationData);
        log.error("ack: {}", ack);
        if (!ack) {
            log.error("异常处理....");
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("return exchange: {}, routingKey: {}, replyCode: {}, replyText: {}", message, routingKey, replyCode, replyText);
    }
}
